package com.hw.kafkaThread03;

import net.sf.jsqlparser.statement.select.Offset;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.util.List;
import java.util.Map;

public class MetricParser extends Thread {
    public final ConsumerRecords<String, String> records;
    public Map<TopicPartition, OffsetAndMetadata> offsets;
    public MetricParser(ConsumerRecords<String, String> records, Map<TopicPartition, OffsetAndMetadata> offsets) {
        this.records = records;
        this.offsets = offsets;
    }
    @Override
    public void run() {
        for (TopicPartition tp : records.partitions()) {
            List<ConsumerRecord<String, String>> tpRecords = records.records(tp);
            long lastOffset = tpRecords.get(tpRecords.size() - 1).offset();
            synchronized (offsets) {
                if (!offsets.containsKey(tp)) {
                    offsets.put(tp, new OffsetAndMetadata(lastOffset + 1));
                } else {
                    long currOffset = offsets.get(tp).offset();
                    if (currOffset < lastOffset + 1) {
                        offsets.put(tp, new OffsetAndMetadata(lastOffset + 1));
                    }
                }
            }
        }
    }
}
